package com.wjw.redis.stream.config;

import com.wjw.redis.stream.constant.CacheConstant;
import com.wjw.redis.stream.model.UserInfo;
import com.wjw.redis.stream.util.RedisUtil;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.data.redis.connection.RedisConnectionFactory;
import org.springframework.data.redis.connection.stream.Consumer;
import org.springframework.data.redis.connection.stream.ObjectRecord;
import org.springframework.data.redis.connection.stream.ReadOffset;
import org.springframework.data.redis.connection.stream.StreamInfo;
import org.springframework.data.redis.connection.stream.StreamOffset;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.hash.ObjectHashMapper;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.stream.StreamMessageListenerContainer;
import org.springframework.data.redis.stream.Subscription;

import javax.annotation.PostConstruct;
import java.time.Duration;
import java.util.HashMap;
import java.util.List;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.stream.Collectors;

/**
 * redis stream配置
 */
@Configuration
public class RedisUserInfoStreamConfig {
    @Autowired
    private RedisUtil redisUtil;
    private final static String[] GROUPS = {"group-c"};

    @Autowired
    private StringRedisTemplate redisTemplate;

    /**
     * 初始化stream和group。否则启动时会报异常
     */
    @PostConstruct
    public void init() {
        boolean hasKey = redisUtil.hasKey(CacheConstant.REDIS_STEAM_02);
        if (!hasKey) {
            HashMap<String, String> map = new HashMap<>();
            map.put("id", CacheConstant.REDIS_STEAM_02);
            map.put("name", "CacheConstant");
            redisUtil.addMap(CacheConstant.REDIS_STEAM_02, map);
        }
        StreamInfo.XInfoGroups groups = redisUtil.queryGroups(CacheConstant.REDIS_STEAM_02);
        List<String> groupNames = groups.stream().map(StreamInfo.XInfoGroup::groupName).collect(Collectors.toList());
        for (String group : GROUPS) {
            if (!groupNames.contains(group)) {
                redisUtil.createGroup(CacheConstant.REDIS_STEAM_02, group);
            }
        }
    }

    @Bean
    public StreamMessageListenerContainer.StreamMessageListenerContainerOptions<String, ObjectRecord<String, UserInfo>> userInfoStreamMessageListenerContainerOption() {

        AtomicInteger index = new AtomicInteger(1);
        int processors = Runtime.getRuntime().availableProcessors();
        ThreadPoolExecutor executor = new ThreadPoolExecutor(processors, processors, 0, TimeUnit.SECONDS,
            new LinkedBlockingDeque<>(), r -> {
            Thread thread = new Thread(r);
            thread.setName("async-stream-consumer-" + index.getAndIncrement());
            thread.setDaemon(true);
            return thread;
        });

        return StreamMessageListenerContainer.StreamMessageListenerContainerOptions
            .builder()
            // 一次最多获取多少条消息
            .batchSize(5)
            // 运行 Stream 的 poll task
            .executor(executor)
            // 可以理解为 Stream Key 的序列化方式
            .keySerializer(RedisSerializer.string())
            // 可以理解为 Stream 后方的字段的 key 的序列化方式
            .hashKeySerializer(RedisSerializer.string())
            // 可以理解为 Stream 后方的字段的 value 的序列化方式
            .hashValueSerializer(RedisSerializer.string())//不设置可能会乱码，也可以在redisConfig中设置
            // Stream 中没有消息时，阻塞多长时间，需要比 `spring.redis.timeout` 的时间小，否则会报超时异常
            .pollTimeout(Duration.ofSeconds(1))
            // ObjectRecord 时，将 对象的 filed 和 value 转换成一个 Map 比如：将Book对象转换成map
            .objectMapper(new ObjectHashMapper())
            // 将发送到Stream中的Record转换成ObjectRecord，转换成具体的类型是这个地方指定的类型
            .targetType(UserInfo.class)
            // 获取消息的过程或获取到消息给具体的消息者处理的过程中，发生了异常的处理
            .errorHandler(new StreamErrorHandler())
            .build();
    }

    @Bean
    public StreamMessageListenerContainer<String, ObjectRecord<String, UserInfo>> userInfoStreamMessageListenerContainer(RedisConnectionFactory factory) {
        StreamMessageListenerContainer<String, ObjectRecord<String, UserInfo>> listenerContainer = StreamMessageListenerContainer.create(factory, userInfoStreamMessageListenerContainerOption());
        listenerContainer.start();
        return listenerContainer;
    }


    @Bean
    public Subscription subscriptionUserInfo(StreamMessageListenerContainer<String, ObjectRecord<String, UserInfo>> streamMessageListenerContainer) {
        // 独立消费
        String streamKey = CacheConstant.REDIS_STEAM_02;
        return streamMessageListenerContainer.receive(Consumer.from("group-c", "consumer-c"),
            StreamOffset.create(streamKey, ReadOffset.lastConsumed()), new UserInfoStreamMessageListener(redisUtil));
        //ReadOffset.latest() –阅读最新消息。
        //ReadOffset.from(…) –在特定消息ID之后阅读。
        //ReadOffset.lastConsumed() –在最后消耗的消息ID之后读取（仅针对消费者组）
    }
}
